package com.sdp.irrigation.server.netty.handle;

import java.util.LinkedHashMap;
import java.util.Map;

import org.springframework.kafka.support.SendResult;
import org.springframework.util.StringUtils;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import com.alibaba.fastjson.JSONObject;
import com.sdp.irrigation.kafka.producer.KafkaProducer;
import com.sdp.irrigation.kafka.property.ServerProperty;
import com.sdp.irrigation.kafka.utils.ProtocolUtils;
import com.sdp.irrigation.protocol.CommPackage;
import com.sdp.irrigation.server.netty.channel.IChannelManager;
import com.sdp.irrigation.server.netty.response.IResponseHandle;
import com.sdp.irrigation.utils.CommConstants;
import com.sdp.irrigation.utils.JsonUtils;
import com.sdp.irrigation.utils.ProtocolTypeEnum;
import com.sdp.irrigation.utils.ServerResponseEnum;

import cn.hutool.core.codec.Base64;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Sharable
public class ServerHandler extends SimpleChannelInboundHandler<String> {
	
	private IChannelManager channelService;
	
	private Map<String, String> eqMap;
	
	private IResponseHandle jsonResponseHandle;
	
	private ServerProperty serverProperty;
	
	private KafkaProducer kafkaProducer;

	public ServerHandler(IChannelManager channelService, Map<String, String> eqMap, IResponseHandle jsonResponseHandle, 
			KafkaProducer kafkaProducer, ServerProperty serverProperty) {
		this.channelService = channelService;
		this.eqMap = eqMap;
		this.jsonResponseHandle = jsonResponseHandle;
		this.kafkaProducer = kafkaProducer;
		this.serverProperty = serverProperty;
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		channelService.initChannel(ctx.channel());
//		ChannelSession channelSession = channelService.getSession(ctx.channel().id());
//		String telNo = "";
//		if (channelSession != null) {
//			telNo = channelSession.getTelNo();
//		}
		log.info("###########channelActive回调，客户端建立连接, 客户端ip为：[{}], channelId: [{}]", ctx.channel().remoteAddress(), ctx.channel().id());
		super.channelActive(ctx);
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		log.error("###########exceptionCaught错误回调，客户端断开连接, 客户端ip为：[{}], channelId: [{}]", ctx.channel().remoteAddress(), ctx.channel().id());
		ctx.close();
		super.exceptionCaught(ctx, cause);
	}
	
	@Override
	public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
		log.debug("###########channelRegistered回调，客户端注册成功，客户端ip为：[{}]", ctx.channel().remoteAddress());
		super.channelRegistered(ctx);
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		//在ChannelManager中处理连接的删除
//		ChannelSession channelSession = channelService.getSession(ctx.channel().id());
//		String telNo = "";
//		if (channelSession != null) {
//			telNo = channelSession.getTelNo();
//		}
//		log.debug("###########channelInactive回调，客户端连接断开,客户端设备编号为： [{}], 客户端ip为：[{}], channelId: [{}]", telNo, ctx.channel().remoteAddress(), ctx.channel().id().toString());
//		if (ctx.channel() != null) {
//			boolean flag = ctx.channel().isOpen();
//			ctx.close();
//		}
		log.info("###########channelInactive回调，客户端连接断开,客户端ip为：[{}], channelId: [{}]", ctx.channel().remoteAddress(), ctx.channel().id().toString());
		super.channelInactive(ctx);
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
		CommPackage packageData = JsonUtils.parseBean(msg, CommPackage.class);
		boolean isSendKafka = false;
		if (packageData != null) {
			String devSn = "";
			if (ProtocolTypeEnum.LOGIN.getCode().equals(packageData.getType())) {
				// 初始化通道session
				devSn = CommConstants.getLoginCommPackageDevSn(packageData);
				if (!StringUtils.isEmpty(devSn)) {
					if (channelService.isChannelExist(ctx.channel().id())) {
						channelService.addMap(devSn, ctx.channel());
					} else {
						log.info("Channel通道不存在，新增-------------");
						channelService.initChannel(ctx.channel());
						channelService.addMap(devSn, ctx.channel());
					}
					
					
//					if (channelService.get(devSn) != null) {
//						Channel oldChannel = channelService.get(devSn);
//						ChannelSession channelSession = channelService.getSession(oldChannel.id());
//						log.info("Channel通道状态: {}, 设备编号为：[{}]", oldChannel.isActive(), channelSession.getTelNo());
//						oldChannel.close();
//					}
//					channelService.addMap(devSn, ctx.channel());
					//注册应答
					CommPackage message = new CommPackage();
					message.setPakSn(CommConstants.generatePakSn());
					message.setCode(ServerResponseEnum.LOGIN_SUCCESS.getCode());
					JSONObject rtn = new JSONObject();
					rtn.put("devSn", devSn);
					message.setRtn(rtn);
					
					jsonResponseHandle.sendMessage(devSn, message);
					isSendKafka = true;
				}
			} else if (ProtocolTypeEnum.HEART.getCode().equals(packageData.getType())) {
				devSn = CommConstants.getHeartCommPackageDevSn(packageData);
				if (!StringUtils.isEmpty(devSn)) {
					if (channelService.isChannelExist(ctx.channel().id())) {
						channelService.addMap(devSn, ctx.channel());
					} else {
						log.info("Channel通道不存在，新增-------------");
						channelService.initChannel(ctx.channel());
						channelService.addMap(devSn, ctx.channel());
					}
					//心跳应答
					Map<String, Object> map = new LinkedHashMap<>();
					//LocalDateUtil.localDateTime2String(LocalDateTime.now(), CommConstants.TIME_FORMATTER)
					map.put("pakSn", CommConstants.generatePakSn());
					map.put("code", ServerResponseEnum.HEART_RESPONSE.getCode());
					map.put("devSn", devSn);
//					CommPackage message = new CommPackage();
//					message.setPakSn(String.valueOf(new Date().getTime()));
//					message.setCode(ServerResponseEnum.LOGIN_SUCCESS.getCode());
//					message.setDevSn(devSn);
					
					jsonResponseHandle.sendMessage(devSn, map);
					isSendKafka = true;
				}
			} else if (ProtocolTypeEnum.REPORT.getCode().equals(packageData.getType())) {
				devSn = CommConstants.getReportCommPackageDevSn(packageData);
				if (!StringUtils.isEmpty(devSn)) {
					if (channelService.isChannelExist(ctx.channel().id())) {
						channelService.addMap(devSn, ctx.channel());
					} else {
						log.info("Channel通道不存在，新增-------------");
						channelService.initChannel(ctx.channel());
						channelService.addMap(devSn, ctx.channel());
					}
					//设备帧应答
					Map<String, Object> map = new LinkedHashMap<>();
					map.put("pakSn", CommConstants.generatePakSn());
					map.put("code", ServerResponseEnum.DEV_RESPONSE.getCode());
					map.put("devSn", devSn);
//					CommPackage message = new CommPackage();
//					message.setPakSn(String.valueOf(new Date().getTime()));
//					message.setCode(ServerResponseEnum.DEV_RESPONSE.getCode());
//					message.setDevSn(devSn);
					jsonResponseHandle.sendMessage(devSn, map);
					isSendKafka = true;
				}
			} else if (ProtocolTypeEnum.USER.getCode().equals(packageData.getType())) {
				devSn = CommConstants.getUserCommPackageDevSn(packageData);
				if (!StringUtils.isEmpty(devSn)) {
					if (channelService.isChannelExist(ctx.channel().id())) {
						channelService.addMap(devSn, ctx.channel());
					} else {
						log.info("Channel通道不存在，新增-------------");
						channelService.initChannel(ctx.channel());
						channelService.addMap(devSn, ctx.channel());
					}
					//用户帧应答,放在业务端处理后再消费应答消息
//					Map<String, Object> map = new LinkedHashMap<>();
//					map.put("pakSn", CommConstants.generatePakSn());
//					map.put("code", ServerResponseEnum.USER_RESPONSE.getCode());
//					map.put("devSn", devSn);
//					jsonResponseHandle.sendMessage(devSn, map);
					isSendKafka = true;
				}

			} else if (ProtocolTypeEnum.ALARM.getCode().equals(packageData.getType())) {
				devSn = CommConstants.getAlarmCommPackageDevSn(packageData);
				if (!StringUtils.isEmpty(devSn)) {
					if (channelService.isChannelExist(ctx.channel().id())) {
						channelService.addMap(devSn, ctx.channel());
					} else {
						log.info("Channel通道不存在，新增-------------");
						channelService.initChannel(ctx.channel());
						channelService.addMap(devSn, ctx.channel());
					}
					//报警应答
					CommPackage message = new CommPackage();
					message.setPakSn(CommConstants.generatePakSn());
					message.setCode(ServerResponseEnum.ALARM_RESPONSE.getCode());
					message.setDevSn(devSn);
					jsonResponseHandle.sendMessage(devSn, message);
					isSendKafka = true;
				}
			} else if (ProtocolTypeEnum.QUERY_BALANCE.getCode().equals(packageData.getType())) {
				devSn = CommConstants.getQueryBalanceCommPackageDevSn(packageData);
				if (!StringUtils.isEmpty(devSn)) {
					if (channelService.isChannelExist(ctx.channel().id())) {
						channelService.addMap(devSn, ctx.channel());
					} else {
						log.info("Channel通道不存在，新增-------------");
						channelService.initChannel(ctx.channel());
						channelService.addMap(devSn, ctx.channel());
					}
					
					//这里不能直接应答，需要等消费端查询远程http接口
					isSendKafka = true;
				}
			} else if (ProtocolTypeEnum.VERSION.getCode().equals(packageData.getType())) {
				devSn = CommConstants.getVersionCommPackageDevSn(packageData);
				if (!StringUtils.isEmpty(devSn)) {
					if (channelService.isChannelExist(ctx.channel().id())) {
						channelService.addMap(devSn, ctx.channel());
					} else {
						log.info("Channel通道不存在，新增-------------");
						channelService.initChannel(ctx.channel());
						channelService.addMap(devSn, ctx.channel());
					}
					
					isSendKafka = true;
				}
			} else if (ProtocolTypeEnum.PARAM_SET.getCode().equals(packageData.getType())) {
				devSn = CommConstants.getParamSetCommPackageDevSn(packageData);
				if (!StringUtils.isEmpty(devSn)) {
					if (channelService.isChannelExist(ctx.channel().id())) {
						channelService.addMap(devSn, ctx.channel());
					} else {
						log.info("Channel通道不存在，新增-------------");
						channelService.initChannel(ctx.channel());
						channelService.addMap(devSn, ctx.channel());
					}
					
					isSendKafka = true;
				}
			}
			
			
//			if ("11041100".equals(devSn)) {
//				log.info("上行消息： {}", msg);
//			}
			
			
			if (isSendKafka) {
				String topic = ProtocolUtils.getUpMessageTopic(packageData.getType(), serverProperty.getUpTopicPrefix());
				
				ListenableFuture<SendResult<String, String>> future = kafkaProducer.getKafkaTemplate().send(topic, 0, null, Base64.encode(msg));
				future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
		            @Override
		            public void onFailure(Throwable ex) {
		            	log.error("kafka sendMessage error, ex = {}, topic = {}, data = {}", ex.getMessage(), topic, msg);
		            }

		            @Override
		            public void onSuccess(SendResult<String, String> result) {
		            	//logger.debug("kafka sendMessage success topic = {}, data = {}", topic, msg);
		            }
		        });
			}
			
		}
	}
	
//	private void sendKafka(PackageMessage<IJtPackage<?>> message) {
//		final String topic = StringUtil.isNullOrEmpty(message.getCustomTopic()) ? BizToolUtils.getKafkaTopic(kafkaProperty.getProduceLogPrefix(), message.getData().getMessageId()) : message.getCustomTopic();
//		int partitionCount = (kafkaProperty.getProduceDefinePartition() != null && kafkaProperty.getProduceDefinePartition().containsKey(message.getData().getMessageId())) ? kafkaProperty.getProduceDefinePartition().get(message.getData().getMessageId()) : kafkaProperty.getProduceDefaultPartition();
//		ListenableFuture<SendResult<String, String>> future = null;
//		String msg = JsonUtils.toJSONString(message);
//		if (partitionCount == 1) {
//			future = kafkaProducer.getKafkaTemplate().send(topic, 0, null, msg);
//		} else {
//			int partition = Math.abs(message.getData().getTelNo().hashCode()) % partitionCount;
//			future = kafkaProducer.getKafkaTemplate().send(topic, partition, null, msg);
//		}
//		future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
//            @Override
//            public void onFailure(Throwable ex) {
//            	logger.error("kafka sendMessage error, ex = {}, topic = {}, data = {}", ex.getMessage(), topic, msg);
//            }
//
//            @Override
//            public void onSuccess(SendResult<String, String> result) {
//            	//logger.debug("kafka sendMessage success topic = {}, data = {}", topic, msg);
//            }
//        });
//	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent)evt;
            if (event.state()== IdleState.READER_IDLE){
            	log.debug("服务端主动断开连接，连接的session信息为：[{}]",
    					JsonUtils.toJSONString(channelService.getSession(ctx.channel().id())));
                ctx.channel().close();
            }
        }else {
            super.userEventTriggered(ctx,evt);
        }

		
//		if (IdleStateEvent.class.isAssignableFrom(evt.getClass())) {
//			logger.debug("服务端主动断开连接，连接的session信息为：[{}]",
//					JsonUtils.toJSONString(channelService.getSession(ctx.channel().id())));
//			ctx.close();
//		}
		// super.userEventTriggered(ctx, evt);
	}
	
	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		// System.out.println("==============channel-read-complete==============");
		ctx.flush();
	}

}
